/*  
	
	prodcon.c -- modified 07-prod-cons.c

	By: Josh Boroff, Ben Gentry, and Adam Wilford

	Date: October 5, 2012

*/
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>
#include <unistd.h>
#include "queue_a.h"


#define BUFFER_SIZE 5
#define NUMTHREADS 2
int idcnt = 1;
int signal_sent = 0;//signal test for creation of buffer
pthread_mutex_t cregion_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t pregion_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t request_notify = PTHREAD_COND_INITIALIZER;
int threadCount = 0; 

/* Active threads linked list */
typedef struct threads_node {
	struct threads_node * next;
	pthread_t threadId;
}ThreadNode;

/* Token struct with active thread pointer */
typedef struct token_struct {
	struct token_struct * next;
	int token;
	int activeThreadCount;
	ThreadNode* head;
} Token;

/* One of these per stream.
   Holds:  the mutex lock and notifier condition variables
           a buffer of tokens taken from a producer
           a structure with information regarding the processing of tokens 
           an identity
*/
typedef struct stream_struct {
   struct stream_struct *next;
   pthread_mutex_t lock;
   pthread_cond_t notifier;
   queue buffer;               /* a buffer of values of any type      */
   void *args;                 /* arguments for the producing stream */
   int id;                     /* identity of this stream            */
} Stream;

/* prod: linked list of streams that this producer consumes tokens from
   self: the producer's output stream */
typedef struct {
   Stream *self, *prod;
} Args;

ThreadNode * threadHead;
Token * tokenHead;

/* return 'value' which should have a value from a call to put */
Token *get(void *stream) {
   pthread_mutex_lock(&cregion_lock);
	printf("sent signal1\n");
   pthread_cond_signal(&request_notify);
   signal_sent = 1;
   Token *ret;  /* needed to take save a value from the critical section */
   queue *q = &((Stream*)stream)->buffer;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);    /* lock other threads out of this section    */
   if (isEmpty(q)) {             /* if nothing in the buffer, wait and open   */
      printf("sent signal2\n");
      pthread_cond_signal(&request_notify);
      pthread_cond_wait(notifier, lock);     /* the section to other threads */
      }
	/*
	// Attempted creating conditions for all consumers to see each token before dequeue

	Token * t = (Token*)(getFirst(q));
	ThreadNode* temp = t->head;
	pthread_t tid = pthread_self();
	pthread_t PTHREAD_NULL;
	
	int found = 0;
	int i,ruined_threads;
	for(i=0;i<t->activeThreadCount;i++) {
		if(pthread_equal(temp->threadId, PTHREAD_NULL)){
			temp = temp->next;
			ruined_threads++;
		}
		if(pthread_equal(tid, temp->threadId)){
			temp->threadId = PTHREAD_NULL;
			found = 1;
			t->activeThreadCount--;
		}
	}
	if(!found && ruined_threads == t->activeThreadCount){
		ret = dequeue(q);
		free(ret);
	}
	ret = t;
	dequeue(q);
	*/
  	ret = dequeue(q);            /* take the next token from the buffer       */
   pthread_cond_signal(notifier);  /* if producer is waiting to add a token  */
   pthread_mutex_unlock(lock);     /* wake it up and unlock the section      */
   pthread_mutex_unlock(&cregion_lock);
   return ret;
}

/* 'value' is the value to move to the consumer */
void put(void *stream, Token *value) {
	printf("Put value: %d\n", value->token);	
	pthread_mutex_lock(&pregion_lock);
   printf("start wait\n");
   if(signal_sent == 0)
       pthread_cond_wait(&request_notify,&pregion_lock);
   signal_sent = 0;
   int j;
   queue *q = &((Stream*)stream)->buffer;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);      /* lock the section */
   if (nelem(q) >= BUFFER_SIZE)   /* if buffer is full, cause the thread to */
      pthread_cond_wait(notifier, lock);  /* wait - unlock the section      */
   enqueue(q,value);              /* add the 'value' token to the buffer    */
   pthread_cond_signal(notifier); /* wake up a sleeping consumer, if any    */
   pthread_mutex_unlock(lock);    /* unlock the section */
   pthread_mutex_unlock(&pregion_lock);
   return;
}

/* Put 1,2,3,4,5... into the self stream */
void *successor (void *streams) {
   Stream *self = ((Args*)streams)->self;
   int id = ((Args*)streams)->self->id;
   int i;
	Token *value;
   
   for (i=1 ; ; i++) {
      printf("Successor(%d): sending %d\n", id, i);
      value = (Token*)malloc(sizeof(Token));
      value->token = i;
		// copies master thread linked list for each individual token
		ThreadNode* temp = threadHead;
		value->head  = malloc(sizeof(ThreadNode));
		value->head->threadId = temp->threadId;
		value->head->next = NULL;
		ThreadNode* temp1 = value->head;
		while(temp->next != NULL) {
			temp = temp->next;
			temp1->next = malloc(sizeof(ThreadNode));
			temp1 = temp1->next;
			temp1->threadId = temp->threadId;
			temp1->next = NULL;
		}
		
      value->activeThreadCount = threadCount;
		put(self, value);
      printf("Successor(%d): sent %d, buf_sz=%d\n",
             id, i, nelem(&self->buffer));
   }
   pthread_exit(NULL);
}

/* Final consumer in the network */
void *consumer (void *streams) {
	
	// initialize thread linked list if empty
	if(threadHead == NULL) {
		threadHead = malloc(sizeof(ThreadNode));
		threadHead->threadId = pthread_self();
		threadHead->next = NULL;
	}
	// populates the thread linked list
	else {
		ThreadNode* temp = threadHead;
		pthread_t threadId = pthread_self();
		int present = 0;
		while(temp->next != NULL){
			if(temp->threadId == threadId)
				present = 1;
		}
		if(!present){
			ThreadNode* temp1 = malloc(sizeof(ThreadNode));
			temp1->threadId = threadId;
			temp1->next = NULL;
			temp->next = temp1;
		}
	}
		
	Stream *prod = ((Args*)streams)->prod;
   
   int i;
	Token* value;
   
   for (i=0 ; i < 10 ; i++) {
		value = get(prod); 
      printf("\t\t\t\t\t\tConsumer(%d): got %d\n",pthread_self(), value->token);
		free(value); // with commented code in get function, uncomment this
   }
   
   pthread_exit(NULL);
}

/* initialize streams - see also queue_a.h and queue_a.c */
void init_stream (Args *args, Stream *self, void *data) {
   if (self != NULL) {
      self->next = NULL;
      self->args = data;
      self->id = idcnt++;
      init_queue(&self->buffer);
      pthread_mutex_init(&self->lock, NULL);
      pthread_cond_init (&self->notifier, NULL);
   }
   args->self = self;
   args->prod = NULL;
}

/* free allocated space in the queue - see queue_a.h and queue_a.c */
void kill_stream(Stream *stream) { destroy_queue(&stream->buffer); }

/* puts an initialized stream object onto the end of a stream's input list */
void connect (Args *arg, Stream *s) {  
   s->next = arg->prod;
   arg->prod = s;
	threadCount++; // increment active thread count (global)
}

void disconnect (Args *arg, Stream *s) {  
   arg->prod = NULL;
	threadCount--;
}

int main () {
   pthread_t s1, c1, c2;
   pthread_t tid1;
	Stream suc1;
   Args suc1_args, cons_args,cons2_args;
   pthread_attr_t attr;
	threadHead = NULL; // beginning of thread linked list
	tokenHead = NULL; // beginning of token liked list

   init_stream(&suc1_args, &suc1, NULL);   /* initialize a successor stream */
   init_stream(&cons_args, NULL, NULL);    /* initialize a consumer stream */
   connect(&cons_args, &suc1);              /* connect to a merge stream */
   init_stream(&cons2_args, NULL, NULL);
   connect(&cons2_args, &suc1);
   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
   pthread_create(&s1, &attr, successor, (void*)&suc1_args);
	
	// create multiple consumer threads
	pthread_create(&c1, &attr, consumer, (void*)&cons_args);  
   pthread_create(&c2, &attr, consumer, (void*)&cons2_args);

	pthread_join(c1, NULL);    /* cancel all running threads when the */
   pthread_join(c2, NULL);
                              /* is finished consuming */
   pthread_cancel(s1);

   kill_stream(&suc1);

   pthread_exit(NULL);
}
   
